---
title: "ACP"
icon: "server"
---

[Agent Communication Protocol (ACP)](https://agentcommunicationprotocol.dev/) is a standard for agent-to-agent communication, allowing different AI agents to interact regardless of how they’re built. This agent works with any ACP-compliant service.

---

### Agent prerequisites

- **[Agent Stack](https://beeai.dev/)** installed and running locally
- **BeeAI Framework** installed with `pip install beeai-framework`
- **BeeAI Framework extension for ACP** installed with `pip install 'beeai-framework[acp]'`


### ACP Agent

When to use ACP instead of [Agent Stack Integration](./agent-stack.mdx)?

- You're connecting to your own custom ACP server
- You're developing a multi-agent system where agents communicate via ACP
- You're integrating with a third-party ACP-compliant service that isn't the Agent Stack

<CodeGroup>

	{/* <!-- embedme python/examples/agents/providers/acp.py --> */}
	```py Python [expandable]
	import asyncio
	import sys
	import traceback
	
	from beeai_framework.adapters.acp.agents import ACPAgent
	from beeai_framework.errors import FrameworkError
	from beeai_framework.memory.unconstrained_memory import UnconstrainedMemory
	from examples.helpers.io import ConsoleReader
	
	
	async def main() -> None:
	    reader = ConsoleReader()
	
	    agent = ACPAgent(agent_name="chat", url="http://127.0.0.1:8001", memory=UnconstrainedMemory())
	    for prompt in reader:
	        # Run the agent and observe events
	        response = await agent.run(prompt).on(
	            "update",
	            lambda data, event: (reader.write("Agent 🤖 (debug) : ", data)),
	        )
	
	        reader.write("Agent 🤖 : ", response.last_message.text)
	
	
	if __name__ == "__main__":
	    try:
	        asyncio.run(main())
	    except FrameworkError as e:
	        traceback.print_exc()
	        sys.exit(e.explain())
	
	```

	{/* <!-- embedme typescript/examples/agents/providers/acp.ts --> */}
	```ts TypeScript [expandable]
	import "dotenv/config.js";
	import { ACPAgent } from "beeai-framework/adapters/acp/agents/agent";
	import { createConsoleReader } from "examples/helpers/io.js";
	import { FrameworkError } from "beeai-framework/errors";
	import { TokenMemory } from "beeai-framework/memory/tokenMemory";
	
	const agentName = "chat";
	
	const agent = new ACPAgent({
	  url: "http://127.0.0.1:8000",
	  agentName,
	  memory: new TokenMemory(),
	});
	
	const reader = createConsoleReader();
	
	try {
	  for await (const { prompt } of reader) {
	    const result = await agent.run({ input: prompt }).observe((emitter) => {
	      emitter.on("update", (data) => {
	        reader.write(`Agent (received progress) 🤖 : `, JSON.stringify(data.value, null, 2));
	      });
	      emitter.on("error", (data) => {
	        reader.write(`Agent (error) 🤖 : `, data.message);
	      });
	    });
	
	    reader.write(`Agent (${agentName}) 🤖 : `, result.result.text);
	  }
	} catch (error) {
	  reader.write("Agent (error)  🤖", FrameworkError.ensure(error).dump());
	}
	
	```

</CodeGroup>

The availability of ACP agents depends on the server you're connecting to. You can check which agents are available by using the check_agent_exists method:

<CodeGroup>

	```py Python [expandable]
	try:
	  await agent.check_agent_exists()
	  print("Agent exists and is available")
	except AgentError as e:
	  print(f"Agent not available: {e.message}")
	```

	```ts TypeScript [expandable]
	try {
	  await agent.checkAgentExists();
	  console.log("Agent exists and is available");
	} catch (e) {
	  console.error(`Agent not available: ${e.message}`);
	}
	```

</CodeGroup>

If you need to create your own ACP server with custom agents, BeeAI framework provides the AcpServer class.

### ACP Server

Basic example:

{/* <!-- embedme python/examples/serve/acp.py --> */}
```py Python [expandable]
from beeai_framework.adapters.acp import ACPServer, ACPServerConfig
from beeai_framework.agents.requirement import RequirementAgent
from beeai_framework.backend import ChatModel
from beeai_framework.memory import UnconstrainedMemory
from beeai_framework.tools.search.duckduckgo import DuckDuckGoSearchTool
from beeai_framework.tools.weather import OpenMeteoTool


def main() -> None:
    llm = ChatModel.from_name("ollama:granite4:micro")
    agent = RequirementAgent(
        llm=llm,
        tools=[DuckDuckGoSearchTool(), OpenMeteoTool()],
        memory=UnconstrainedMemory(),
        # specify the agent's name and other metadata
        name="chat",
        description="A simple agent",
    )

    # Register the agent with the ACP server and run the HTTP server
    # For the ToolCallingAgent and ReActAgent, we don't need to specify ACPAgent factory method
    # because they are already registered in the ACPServer
    ACPServer(config=ACPServerConfig(port=8001)).register(agent, tags=["example"]).serve()


if __name__ == "__main__":
    main()

```

**Custom agent example:**

{/* <!-- embedme python/examples/serve/acp_with_custom_agent.py --> */}
```py Python [expandable]
import sys
import traceback
from collections.abc import AsyncGenerator
from typing import Unpack

import acp_sdk.models as acp_models
import acp_sdk.server.context as acp_context
import acp_sdk.server.types as acp_types

from beeai_framework.adapters.acp import ACPServer
from beeai_framework.adapters.acp.serve._utils import acp_msgs_to_framework_msgs
from beeai_framework.adapters.acp.serve.agent import ACPServerAgent
from beeai_framework.adapters.acp.serve.server import ACPServerMetadata, to_acp_agent_metadata
from beeai_framework.agents import AgentOptions, AgentOutput, BaseAgent
from beeai_framework.backend.message import AnyMessage, AssistantMessage, UserMessage
from beeai_framework.emitter.emitter import Emitter
from beeai_framework.errors import FrameworkError
from beeai_framework.memory import UnconstrainedMemory
from beeai_framework.memory.base_memory import BaseMemory
from beeai_framework.runnable import runnable_entry


# This is a simple echo agent that echoes back the last message it received.
class EchoAgent(BaseAgent):
    memory: BaseMemory

    def __init__(self, memory: BaseMemory) -> None:
        super().__init__()
        self.memory = memory

    def _create_emitter(self) -> Emitter:
        return Emitter.root().child(
            namespace=["agent", "custom"],
            creator=self,
        )

    @runnable_entry
    async def run(self, input: str | list[AnyMessage], /, **kwargs: Unpack[AgentOptions]) -> AgentOutput:
        assert self.memory is not None

        if isinstance(input, str):
            await self.memory.add(UserMessage(input))
        elif isinstance(input, list):
            await self.memory.add_many(input)

        text_input = self.memory.messages[-1].text if self.memory.messages else ""
        return AgentOutput(output=[AssistantMessage(text_input)])


def main() -> None:
    # Create a custom agent factory for the EchoAgent
    def agent_factory(agent: EchoAgent, *, metadata: ACPServerMetadata | None = None) -> ACPServerAgent:
        """Factory method to create an ACPAgent from a EchoAgent."""
        if metadata is None:
            metadata = {}

        async def run(
            input: list[acp_models.Message], context: acp_context.Context
        ) -> AsyncGenerator[acp_types.RunYield, acp_types.RunYieldResume]:
            framework_messages = acp_msgs_to_framework_msgs(input)
            response = await agent.run(framework_messages)
            yield acp_models.MessagePart(content=response.last_message.text, role="assistant")  # type: ignore[call-arg]

        # Create an ACPAgent instance with the run function
        return ACPServerAgent(
            fn=run,
            name=metadata.get("name", agent.meta.name),
            description=metadata.get("description", agent.meta.description),
            metadata=to_acp_agent_metadata(metadata),
        )

    # Register the custom agent factory with the ACP server
    ACPServer.register_factory(EchoAgent, agent_factory)
    # Create an instance of the EchoAgent with UnconstrainedMemory
    agent = EchoAgent(memory=UnconstrainedMemory())
    # Register the agent with the ACP server and run the HTTP server
    ACPServer().register(agent, name="echo_agent").serve()


if __name__ == "__main__":
    try:
        main()
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())

# run: beeai agent run echo_agent "Hello"

```

_Source: python/examples/serve/acp_with_custom_agent.py_
